package knowlwdge_graph.knowledge_etl.yago_etl;

import common.dao.postgre.PostGreJDBCDao;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;

/**
 * Created by julianzliu on 4/17/2017.
 */
public class TransformerP2OMultiThread {

    static final org.slf4j.Logger logger = LoggerFactory.getLogger(TransformerP2OMultiThread.class);

    private PostGreJDBCDao postGreJDBCDao;
    private Connection conn;
    private Integer thread_num;

    public TransformerP2OMultiThread(Integer thread_num){
        this.postGreJDBCDao = new PostGreJDBCDao();
        this.conn = this.postGreJDBCDao.getConn();
        this.thread_num = thread_num;
        logger.info("[TransformerP2OMultiThread] thread_num = {}", this.thread_num);
    }


    public int getTotalSize(){
        String sql = String.format("select count(*) as total_size  from yagofacts;");
        logger.info("[getTotalSize] sql = {}", sql);
        Integer total_size = 0;
        try {
            Statement stmt = conn.createStatement();
            ResultSet rs = null;
            rs = stmt.executeQuery(sql);
            while(rs.next()){
                total_size = rs.getInt("total_size");
            }
            rs.close();
            stmt.close();
        } catch (SQLException e){
            e.printStackTrace();
        }
        return total_size;
    }


    public void createThread(int begin, int end, int batch_size){
        (new Thread( new TransformerP2O.TransformerThread(begin, end, batch_size) )).start();
    }

    public int getPerThreadSize(int total_size, int thread_num, int batch_size){
        return ( ( (total_size/thread_num) / batch_size + 1) * batch_size );
    }


    public void createMultiThread(int batch_size){
        // TODO:
        //int total_size = getTotalSize();
        int total_size = 374429460;
        logger.info("[createMultiThread] total_size = {}\n\n", total_size);

        int per_thread_size = getPerThreadSize(total_size,thread_num,batch_size);
        int curr_size = 0;
        int curr_thread = 0;
        while(curr_size < total_size){

            int begin = curr_size;
            int end = Math.min(total_size, curr_size + per_thread_size);
            createThread( begin, end, batch_size );
            curr_thread += 1;
            logger.info("[createMultiThread] curr_thread = {} \n\n ", curr_thread);

            curr_size = end;
        }
    }





    /*
    ******************************************************************
     */
    public static void main(String[] args){

        SchemaConstructer constructer = new SchemaConstructer();
        constructer.createSchema();

        int thread_num = 5;  //五十万
        thread_num = Integer.parseInt(args[0]);
        int batch_size = 10000;  // 一万
        TransformerP2OMultiThread multiThread = new TransformerP2OMultiThread( thread_num );
        multiThread.createMultiThread( batch_size );

    }


}
